package cn.xinfei.xdecision.engine.api;

import cn.xinfei.xdecision.common.utils.util.JsonUtils;
import cn.xinfei.xdecision.model.DecisionPulsarRequest;
import cn.xinfei.xdecision.service.PulsarService;
import com.alibaba.fastjson.JSONObject;
import io.swagger.annotations.ApiOperation;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.ConsumerStats;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;

@Slf4j
@RestController
@RequestMapping("engine/pulsar")
public class XDecisionPulsarApi {

    @Autowired
    private PulsarService pulsarService;

    /**
     * getMessageById
     *
     * @param apiRequest
     * @return
     */
    @RequestMapping(value = "/consumer/getMessageById", method = RequestMethod.POST, produces = "application/json;charset=UTF-8")
    @ApiOperation(value = "getMessageById")
    public Object getMessageById(@RequestBody DecisionPulsarRequest apiRequest) {
        log.info("restart request:{}", JsonUtils.toJsonString(apiRequest));

        return null;
    }

    /**
     * stats
     *
     * @param apiRequest
     * @return
     */
    @RequestMapping(value = "/consumer/stats", method = RequestMethod.POST, produces = "application/json;charset=UTF-8")
    @ApiOperation(value = "stats")
    public ConsumerStats stats(@RequestBody DecisionPulsarRequest apiRequest) {
        log.info("stats request:{}", JsonUtils.toJsonString(apiRequest));
        ConsumerStats stats = pulsarService.stats(apiRequest.getTopic());
        return stats;
    }

    /**
     * 重起消费者
     *
     * @param apiRequest
     * @return
     */
    @RequestMapping(value = "/consumer/restart", method = RequestMethod.POST, produces = "application/json;charset=UTF-8")
    @ApiOperation(value = "重启消费者")
    public Object restart(@RequestBody DecisionPulsarRequest apiRequest) {
        log.info("restart request:{}", JsonUtils.toJsonString(apiRequest));
        pulsarService.restart(apiRequest.getTopic());
        return true;
    }


    @RequestMapping(value = "/consumer/stop", method = RequestMethod.POST, produces = "application/json;charset=UTF-8")
    @ApiOperation(value = "停止消费者")
    public Object stop(@RequestBody DecisionPulsarRequest apiRequest) {
        log.info("stop request:{}", JsonUtils.toJsonString(apiRequest));
        pulsarService.stop(apiRequest.getTopic());
        return true;
    }


}
